<?php
/**
 * Created by PhpStorm.
 * User: chenhong
 * Date: 17/2/16
 * Time: 上午9:06
 * Desc 此队列基于 swoole 和 redis 开发。适用于 fpm 环境
 */
use Swoole\Redis\Server;
date_default_timezone_set('PRC');
include 'Pdo.php';
class Queue
{
    private $serverName = 'Apk unpackage server';
    /**
     * default config
     * @var array
     */
    private $config = [
        'host' => '127.0.0.1',
        'port' => 9501,
        'task_worker_num' => 10,
        'worker_num' => 1,
        'task_max_request' => 1,
        'daemonize' => true,
        'log_file' => './package.log'
    ];

    private $server;

    public function __construct($config = [])
    {
        if (!empty($config)){
            $this->config = array_merge($this->config,$config);
        }

        $server = new Server($this->config['host'],$this->config['port']);
        $this->server = $server;
        $this->server->set($this->config);
        $this->server->setHandler('LPUSH',[$this,'handler']);
        $this->server->on('task',[$this,'task']);
        $this->server->on('finish',[$this,'finish']);
        $this->server->on('start',[$this,'onStart']);
        $this->server->on('managerStart',[$this,'onManageStart']);
        $this->server->on('WorkerStart',[$this,'onWorkerStart']);
        $this->server->start();
    }

    public function onStart()
    {
        $processName = $this->serverName . ' running redis://' . $this->config['host'] . ':' .
            $this->config['port'] . ' time:' . date('Y-m-d H:i:s') . ' master:' . $this->server->master_pid;
        swoole_set_process_name($processName);
    }

    public function onManageStart($server)
    {
        $processName = $this->serverName . ' manager:' . $server->manager_pid;
        swoole_set_process_name($processName);
    }

    public function onWorkerStart($server,$worker_id)
    {
        if ($worker_id >= $server->setting['worker_num'] )
        {
            //worker 进程
            $processName = $this->serverName . ' task worker num:' . $worker_id . ' pid ' . $server->worker_pid;
            swoole_set_process_name($processName);
        } else {
            // task 进程
            //worker 进程
            $processName = $this->serverName . ' worker num:' . $worker_id . ' pid ' . $server->worker_pid;
            swoole_set_process_name($processName);
        }
    }

    //设置 lpush 命令
    public function handler($fd,$data)
    {
        $taskId = $this->server->task($data);
        if ($taskId == false){
            return Server::format(Server::ERROR);
        } else {
            return Server::format(Server::INT,$taskId);
        }
    }

    //处理投递进来的任务
    public function task($serv,$taskId,$workerId,$taskData)
    {
        $pdo =  DB::getInstance([
            'dsn' => 'mysql:dbname=zysdk;host:127.0.0.1',
            'user' => 'root',
            'pass' => 'Jiawangame%90&*$',
            'charset' => 'UTF8',
            'dbname' => 'zysdk'
        ]);
        $data = json_decode($taskData[1],true);
        if (isset($data['is_public']) && $data['is_public'] == true){
            $pdo->setTableName('game_chairman_package');
        } else {
            $pdo->setTableName('game_channel_package');
        }
        $queueKey = $taskData[0];
        $className = ucfirst($queueKey);
        include $className . '.php';
        return $className::parseParam($data,$pdo);
    }

    //任务处理完毕
    public function finish($server,$taskId,$result)
    {
    }
}

$serv = new Queue();